#include "dds_topic_subscribe.h"

#include <stdio.h>
#include <stdlib.h>
#include <iostream>

// #include "dds/dds.h"
// #include "Common.h"
// #include "changan/common_data_types.h"

#include "public_define.h"
#include "data_structure.h"
#include "data_cache.h"

using namespace DcAdr;


DDSTopicSubScribe::DDSTopicSubScribe()
{
}

DDSTopicSubScribe::~DDSTopicSubScribe()
{
}

void on_notify_data_available(int32_t reader, void *arg)
{
    (void)reader;
    (void)arg;
#ifdef BUILD_FOR_C3B
    static dds_return_t rc;
    void *samples[1];
    samples[0] = DDSData_Msg__alloc();
    dds_sample_info_t infos[1];
    rc = dds_take_next(reader, samples, infos);
    if (rc < 0)
    {
        Log_ERROR("dds take next failed!");
    }
    Log_Info("revice topic: ");
    if (infos[0].valid_data)
    {
        DDSData_Msg *request = (DDSData_Msg *)samples[0];
        Log_Info("revice topic: {}", std::string(request->topic));

        ProtoData data;
        data.timestamp = request->timestamp;
        data.topic = std::make_shared<std::string>(request->topic);
        data._length = request->payload._length;
        data.data = std::make_shared<std::string>((char *)request->payload._buffer, data._length);

        DataCache::getInstance().addData(data);
    }
    DDSData_Msg_free(samples, DDS_FREE_ALL);
#endif
}

int DDSTopicSubScribe::init(const std::string &topicStr)
{
    (void)topicStr;
#ifdef BUILD_FOR_C3B
    dds_entity_t participant = -1;
    dds_entity_t topic = -1;
    dds_entity_t subscriber = -1;
    dds_entity_t reader = -1;
    dds_return_t rc = DDS_RETCODE_ERROR;
    dds_qos_t *qos = NULL;
    dds_listener_t *listener = NULL;

    /* 创建一个参与者。 */
    participant = dds_create_participant(0, NULL, NULL);
    if (participant < 0)
    {
        DDS_FATAL("dds_create_participant: %s\n", dds_strretcode(-participant));
    }
    Log_Info("topicStr: {}", topicStr);
    /* 创建一个 Topic 主题 */
    topic = dds_create_topic(participant, &DDSData_Msg_desc, topicStr.c_str(), NULL, NULL);
    if (topic < 0)
    {
        DDS_FATAL("dds_create_topic: %s\n", dds_strretcode(topic));
    }
    const char *Partitions = "ntf";
    const char *subParts[1];
    //指针数组成员指向partition字符串
    subParts[0] = Partitions;
    dds_qos_t *subQos = dds_create_qos();
    /*把partition关联到pubQos中，注意第三个参数是二级指针*/
    dds_qset_partition(subQos, 1, subParts);
    /*Create a subscriber*/
    subscriber = dds_create_subscriber(participant, subQos, NULL);
    if (subscriber < 0)
        Log_ERROR("dds_create_subscriber: {}", std::string(dds_strretcode(-subscriber)));

    /* 创建一个可靠（reliable）的读取者。 */
    qos = dds_create_qos();
    dds_qset_reliability(qos, DDS_RELIABILITY_RELIABLE, DDS_SECS(10));
    listener = dds_create_listener(NULL);

    // bind callback
    dds_lset_data_available(listener, on_notify_data_available);

    reader = dds_create_reader(subscriber, topic, qos, listener);
    if (reader < 0)
        DDS_FATAL("dds_create_reader: %s\n", dds_strretcode(-reader));
    dds_delete_qos(qos);
#endif
    return EXIT_SUCCESS;
}